package com.eshipenet.shipnet.edpersistor

import java.util.Properties

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object EquipmentDataSourceProvider {

  def create(env: StreamExecutionEnvironment): DataStream[EquipmentData] = {
    val props = Map(
      "bootstrap.servers" -> "kafka-server:9092",
      "group.id"          -> "edpersistor.EquipmentDataStream",
    )
    val properties = new Properties()
    props.foreach(kv => properties.setProperty(kv._1, kv._2))
    val topic = "EquipmentData"
    val der   = new RawKafkaMessageDeserializer
    val consumer = new FlinkKafkaConsumer[RawKafkaMessage](
      topic,
      der,
      properties
    )
    consumer.setStartFromLatest()
    val stream = env
      .addSource(consumer)
      .map(msg => EquipmentData.parse(msg.value))
      .filter(_.isSuccess)
      .map(_.get)
    stream
  }

}
